package master

import (
	"mapreduce-go/common"
	"mapreduce-go/schedule"
	"net"
	"sync"
)

type Master struct {
	Address string // master地址

	sync.Mutex // 锁

	NewCond *sync.Cond

	Workers []string // 节点

	JobName string // 任务名称

	Files []string // 输入文件

	NReduce int // reduce分区数量

	Listener net.Listener

	Shutdown chan struct{}

	State []int

	done chan struct{} // 结束标志
}

// 任务调度函数
func Sequential(
	jobName string,
	files []string,
	nReduce int,
	mapF func(string) (res []common.KeyValue),
	reduceF func(string, []string) string) {

	// 执行分配的任务
	// 在MapReduce中任务主要分为Map任务和Reduce任务
	mr := NewMaster("Master", jobName, files, nReduce)
	mr.run(func(phase common.JobPhase) {
		switch phase {
		case common.MapPhase:
			// 执行map任务
			// map任务的调用次数由输入文件的个数决定
			for i, filename := range files {
				common.DoMap(mr.JobName, i, filename, mr.NReduce, mapF)
			}
		case common.ReducePhase:
			for i := 0; i < nReduce; i++ {
				common.DoReduce(mr.JobName, i, common.GetMergeName(mr.JobName, i), len(mr.Files), reduceF)
			}
		}
	}, func() {})
}

// 初始化Master实例
func NewMaster(master, jobName string, files []string, nReduce int) *Master {
	mas := &Master{
		Address:  master,
		JobName:  jobName,
		NReduce:  nReduce,
		Files:    files,
		Shutdown: make(chan struct{}),
		done:     make(chan struct{}),
	}

	mas.NewCond = sync.NewCond(mas)

	return mas
}

func (mr *Master) run(
	schedule func(phase common.JobPhase),
	finish func()) {

	// 执行Map任务
	schedule(common.MapPhase)

	// 执行Reduce任务
	schedule(common.ReducePhase)

	// 合并
	mr.merge()

	finish()
}

// 分布执行MapReduce任务
// 向通过rpc在主服务器注册的worker调度map和reduce任务
func Distribute(
	jobName string,
	files []string,
	nReduce int,
	master string) (mr *Master) {

	mr = NewMaster(master, jobName, files, nReduce)

	mr.StartRpcServer()

	go mr.run(
		func(phase common.JobPhase) {
			ch := make(chan *string, 100)
			go mr.ForwardRegistration(ch)
			schedule.Schedule(mr.JobName, mr.Files, mr.NReduce, phase, ch)
		}, func() {
			// TODO: finish
			mr.done <- struct{}{}
		})

	return
}

func (mr *Master) Done() <-chan struct{} {
	return mr.done
}
